package com.atguigu.tingshu.album.service.impl;
import java.math.BigDecimal;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.common.anno.GuiguCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import com.google.common.collect.Maps;

import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService ;

	@Autowired
	private AlbumStatMapper albumStatMapper ;

	@Autowired
	private KafkaService kafkaService ;

	@Autowired
	private RedisTemplate<String , String> redisTemplate ;

	@Autowired
	private RedissonClient redissonClient ;

	@Transactional
	@Override
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {

		// 保存专辑基本信息到album_info表中
		AlbumInfo albumInfo = new AlbumInfo() ;					// 创建AlbumInfo对象，封装要保存到专辑的数据
		BeanUtils.copyProperties(albumInfoVo , albumInfo);		// 进行对象属性的copy，要求两个对象的属性名称和属性类型必须一致
		albumInfo.setUserId(AuthContextHolder.getUserId());		// 获取当前登录用户的id

		/**
		 * 后期用户访问该接口的时候需要验证用户的登录状态，如果用户登录了，那么此时就可以访问接口。
		 * 验证完用户的登录状态以后，会把当前登录用户信息存储到ThreadLocal中，因此在该接口中要
		 * 获取当前登录用户的id，就可以直接从ThreadLocal中进行获取。
		 */

		// 如何专辑是vip免费或者是付费，那么此时需要给专辑设置一个免费的试听声音集数。
		String payType = albumInfoVo.getPayType();
		if(!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(payType)) {
			albumInfo.setTracksForFree(5);
		}

		// 代表的是审核状态，专辑录入完毕以后，根据需要对专辑数据进行审核（判断专辑的数据中是否包含了非法数据）
		// 审核的时候审核方式：1、人工审核，2、机器审核（需要调用第三方的内容安全服务）
		// 在业务代码中要进行审核：先机器审核  ----->  人工审核
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		albumInfoMapper.insert(albumInfo) ;				// 保存专辑的数据

		// 保存专辑的属性值数据到album_attribute_value中
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();

		/**
		 * stream流中的map方法：对流中的数据类型做转换
		 * lambda表达式：简化匿名内部类的书写
		 */
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
			AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
			albumAttributeValue.setAlbumId(albumInfo.getId());
			albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
			albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
			return albumAttributeValue;
		}).collect(Collectors.toList());
		albumAttributeValueService.saveBatch(albumAttributeValueList) ;

		// 初始化专辑的统计数据到album_stat表
		this.saveAlbumStat(albumInfo.getId() , SystemConstant.ALBUM_STAT_PLAY);
		this.saveAlbumStat(albumInfo.getId() , SystemConstant.ALBUM_STAT_SUBSCRIBE);
		this.saveAlbumStat(albumInfo.getId() , SystemConstant.ALBUM_STAT_BROWSE);
		this.saveAlbumStat(albumInfo.getId() , SystemConstant.ALBUM_STAT_COMMENT);

		// 根据isOpen字段的值进行专辑的上架操作
		String isOpen = albumInfo.getIsOpen();
		if(SystemConstant.ALBUM_IS_OPEN_YES.equals(isOpen)) {
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER , String.valueOf(albumInfo.getId()));
		}

	}

	@Override
	public IPage<AlbumListVo> findUserAlbumPage(Integer pageNo, Integer pageSize, AlbumInfoQuery albumInfoQuery) {

		/**
		 * 所查询到的数据包含专辑的基本信息和专辑的统计数据
		 */
		IPage<AlbumListVo> page = new Page<>(pageNo , pageSize) ;			// 封装分页查询参数

		// 获取当前登录用户的id
		Long userId = AuthContextHolder.getUserId();
		albumInfoQuery.setUserId(userId);

		// 调用albumInfoMapper接口中的自定义条件分页查询方法
		IPage<AlbumListVo> pageList = albumInfoMapper.findUserAlbumPage(page , albumInfoQuery) ;

		// 返回
		return pageList;
	}

	@Transactional
	@Override
	public void removeAlbumInfo(Long id) {

		// 从album_info表中根据专辑的id删除专辑数据
		albumInfoMapper.deleteById(id) ;

		// 从album_attribute_value表中根据专辑的id删除专辑属性值数据
		// delete from album_attribute_value where album_id = ?
		LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>() ;
		albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId , id) ;
		albumAttributeValueService.remove(albumAttributeValueLambdaQueryWrapper) ;

		// 从album_stat表中根据专辑的id删除专辑统计数据
		LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>() ;
		albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId , id) ;
		albumStatMapper.delete(albumStatLambdaQueryWrapper) ;

		// 从ES索引库中删除专辑
		kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER , String.valueOf(id));

	}

	@GuiguCache(
			cacheKey = RedisConstant.ALBUM_INFO_PREFIX + "#{ #params[0] }"
	)
	@Override
	public AlbumInfo getAlbumInfo(Long albumId) {		// 根据专辑的id查询专辑的基本信息(album_info)和专辑的属性值数据(album_attribute_value)
		return albumInfoMapper.getAlbumInfo(albumId);
	}

	/**
	 * 为了提高该接口的数据查询效率，需要使用查询缓存
	 * 查询缓存思想：
	 * 1、先从缓存中查询数据，缓存中如果存在数据直接进行返回
	 * 2、缓存中如果不存在数据，此时再查询数据库，然后把从数据库中查询到的数据返回。但是在返回之前需要将数据存储到缓存中。
	 */
	public AlbumInfo getAlbumInfoRedisson(Long albumId) {		// 根据专辑的id查询专辑的基本信息(album_info)和专辑的属性值数据(album_attribute_value)

		// 从缓存中查询数据
		String albumInfoKey = RedisConstant.ALBUM_INFO_PREFIX + albumId ;
		String albumInfoJSON = redisTemplate.opsForValue().get(albumInfoKey);
		if(!StringUtils.isEmpty(albumInfoJSON)) {
			AlbumInfo albumInfo = JSON.parseObject(albumInfoJSON, AlbumInfo.class);
			log.info("从缓存中查询专辑数据...");
			return albumInfo ;
		}

		// 使用分布式锁解决缓存击穿的问题
		RLock rLock = redissonClient.getLock(RedisConstant.ALBUM_LOCK_SUFFIX + albumId);
		boolean tryLock = rLock.tryLock();
		if(tryLock) {			// 获取到锁，从数据库中查询数据

			log.info(Thread.currentThread().getId() + "该线程获取到了分布式锁，查询数据...");

			try {

				// 从数据库中查询数据
				AlbumInfo albumInfo = albumInfoMapper.getAlbumInfo(albumId);
				redisTemplate.opsForValue().set(albumInfoKey , JSON.toJSONString(albumInfo) , RedisConstant.ALBUM_TIMEOUT , TimeUnit.DAYS);
				log.info("从从数据库中查询到了数据...");

				// 返回查询到的数据
				return albumInfo ;

			}catch (Exception e) {
				e.printStackTrace();
				throw new GuiguException(ResultCodeEnum.DATA_ERROR) ;
			} finally {
				log.info(Thread.currentThread().getId() + "该线程释放了分布式锁...");
				rLock.unlock();
			}


		}else {

			albumInfoJSON = redisTemplate.opsForValue().get(albumInfoKey);
			if(!StringUtils.isEmpty(albumInfoJSON)) {
				AlbumInfo albumInfo = JSON.parseObject(albumInfoJSON, AlbumInfo.class);
				log.info(Thread.currentThread().getId() + "从缓存中查询专辑数据...");
				return albumInfo ;
			}else {
				log.info(Thread.currentThread().getId() + "该线程暂时没有从Redis中查询到数据，进行自旋....");
				return this.getAlbumInfo(albumId) ;		// 进行自旋
			}

		}

	}

	@Transactional
	@Override
	public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {

		// 从缓存中删除专辑的数据
//		String albumInfoKey = RedisConstant.ALBUM_INFO_PREFIX + albumId ;
//		redisTemplate.delete(albumInfoKey) ;

		// 修改专辑的基本数据
		AlbumInfo albumInfo = this.getById(albumId);
		BeanUtils.copyProperties(albumInfoVo , albumInfo);
		this.updateById(albumInfo) ;

		// 修改专辑的属性值数据
		// 根据专辑的id删除之前的属性值数据
		LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>() ;
		albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId , albumId) ;
		albumAttributeValueService.remove(albumAttributeValueLambdaQueryWrapper) ;

		// 新增新的属性值数据
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
			AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
			albumAttributeValue.setAlbumId(albumId);
			albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
			albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
			return albumAttributeValue;
		}).collect(Collectors.toList());
		albumAttributeValueService.saveBatch(albumAttributeValueList) ;

		// 获取isOpen字段值进行判断，
		String isOpen = albumInfo.getIsOpen();
		if(SystemConstant.ALBUM_IS_OPEN_YES.equals(isOpen)) {
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER , String.valueOf(albumId));
		}else {
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER , String.valueOf(albumId));
		}

		// 向Kafka中发送通过Redis的数据消息

		// 延迟删除缓存数据
//		ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(10);
//		scheduledThreadPool.schedule(() -> {
//			redisTemplate.delete(albumInfoKey) ;
//			log.info("延迟任务执行了，从redis中删除了专辑数据....");
//		} , 300 , TimeUnit.MILLISECONDS) ;

	}

	@Override
	public List<AlbumInfo> findAllAlbumList() {

		// 根据当前登录用户的id查询专辑列表数据
		Long userId = AuthContextHolder.getUserId();
		LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>() ;
		albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId , userId) ;
		albumInfoLambdaQueryWrapper.select(AlbumInfo::getId , AlbumInfo::getAlbumTitle) ;  // 指定参数字段: id , album_title

		// sql语句：select id , album_title from album_info where user_id = ?
		List<AlbumInfo> albumInfoList = albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);

		// 返回
		return albumInfoList;
	}

	@GuiguCache(cacheKey = RedisConstant.ALBUM_STAT_PREFIX + "#{ #params[0] }" ,
		enableLock = true ,
		lockName =  RedisConstant.ALBUM_STAT_LOCK_PREFIX + "#{ #params[0] }"
	)
	@Override
	public AlbumStatVo findAlbumStatByAlbumId(Long alumId) {
		return albumStatMapper.findAlbumStatByAlbumId(alumId) ;
	}

	public void saveAlbumStat(Long albumId , String statType) {
		AlbumStat albumStat = new AlbumStat() ;
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statType);
		albumStat.setStatNum(0);
		albumStatMapper.insert(albumStat) ;
	}

//	@PostConstruct
//	public void init() {
//		RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
//		bloomFilter.tryInit(10000000 , 0.00001) ;
//		log.info("布隆过滤器初始化完毕...");
//	}

}
